package com.yeyks.rocketmq.config;

import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.PropertyValueConst;
import com.yeyks.common.constants.rocketmq.TopicAndTag;
import com.yeyks.rocketmq.service.customer.*;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.Properties;

/**
 * @author fuchangshuai
 * @date 2019/08/12 16:32
 */
@Data
@Slf4j
@Component
@ConditionalOnProperty(prefix = "rocket-mq", name = "enable", havingValue = "true")
public class RocketMqConsumer {
    @Autowired
    private RocketMqConfig rocketMqConfig;

    /**
     * 延迟抢单
     *
     * @return
     */
    @Bean(name = "cToWebOrderCheckConsumer")
    public Consumer getCToWebOrderCheckConsumer(CToWebOrderCheckListener cToWebOrderCheckListener) {
        log.info("初始化延迟抢单启动消费者！");
        Properties properties = rocketMqConfig.getProperties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, TopicAndTag.GID_YYKSDELAY_GRABTHEORDERCHECK);
        // 集群订阅方式
        properties.setProperty(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(TopicAndTag.TOPIC_YYKS_DELAY, TopicAndTag.TAG_GRAB_THE_ORDER_CHECK, cToWebOrderCheckListener);
        consumer.start();
        log.info("启动延迟抢单消费者成功！");
        return consumer;
    }

    /**
     * 抢单
     *
     * @return
     */
    @Bean(name = "cToWebOrderConsumer")
    public Consumer getCToWebOrderConsumer(CToWebOrderListener cToWebOrderListener) {
        log.info("初始化抢单启动消费者！");
        // producer 实例配置初始化
        Properties properties = rocketMqConfig.getProperties();
        // 集群订阅方式
        properties.setProperty(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
        properties.setProperty(PropertyKeyConst.GROUP_ID, TopicAndTag.GID_YYKSNORMAL_GRABTHEORDER);
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_GRAB_THE_ORDER, cToWebOrderListener);
        consumer.start();
        log.info("启动抢单消费者成功！");
        return consumer;
    }

    /**
     * 抢单成功
     *
     * @return
     */
    @Bean(name = "cToWebOrderSuccessConsumer")
    public Consumer getCToWebOrderSuccessConsumer(CToWebOrderSuccessListener cToWebOrderSuccessListener) {
        log.info("初始化抢单成功启动消费者！");
        // producer 实例配置初始化
        Properties properties = rocketMqConfig.getProperties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, TopicAndTag.GID_YYKSNORMAL_GRABTHEORDERSUCCESS);
        // 集群订阅方式
        properties.setProperty(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_GRAB_THE_ORDER_SUCCESS, cToWebOrderSuccessListener);
        consumer.start();
        log.info("启动抢单成功消费者成功！");
        return consumer;
    }

    /**
     * 下单30分钟检测
     *
     * @return
     */
    @Bean(name = "orderDelayCheckConsumer")
    public Consumer getOrderDelayCheckConsumer(OrderDelayCheckListener orderDelayCheckListener) {
        log.info("初始化下单30分钟检测启动消费者！");
        Properties properties = rocketMqConfig.getProperties();
        // 集群订阅方式
        properties.setProperty(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
        properties.setProperty(PropertyKeyConst.GROUP_ID, TopicAndTag.GID_YYKSDELAY_CHECKNOTPAY);
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(TopicAndTag.TOPIC_YYKS_DELAY, TopicAndTag.TAG_CHECK_NOT_PAY, orderDelayCheckListener);
        consumer.start();
        log.info("启动下单30分钟检测消费者成功！");
        return consumer;
    }

    /**
     * webSocket监听订单抢单结果
     *
     * @return
     */
    @Bean(name = "webSocketConsumer")
    public Consumer getWebSocketConsumer(WebSocketListener webSocketListener) {
        log.info("初始化webSocket启动消费者！");
        Properties properties = rocketMqConfig.getProperties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, TopicAndTag.GID_YYKSNORMAL_WEBSOCKET);
        // 广播订阅方式
        properties.setProperty(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_WEBSOCKET, webSocketListener);
        consumer.start();
        log.info("启动webSocket消费者成功！");
        return consumer;
    }

    /**
     * webSocket监听订单退款
     *
     * @return
     */
    @Bean(name = "webSocketRefundConsumer")
    public Consumer getWebSocketRefundConsumer(WebSocketOrderRefundListener webSocketOrderRefundListener) {
        log.info("初始化webSocket订单退款启动消费者！");
        Properties properties = rocketMqConfig.getProperties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, TopicAndTag.GID_YYKSNORMAL_REFUNDNOTICE);
        // 集群消费模式
        properties.setProperty(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_REFUND_NOTICE, webSocketOrderRefundListener);
        consumer.start();
        log.info("启动webSocket订单退款消费者成功！");
        return consumer;
    }

    /**
     * 监听订单退款（包括yyks和cn）
     *
     * @return
     */
    @Bean(name = "refundConsumer")
    public Consumer getRefundConsumer(OrderRefundListener orderRefundListener) {
        log.info("初始化订单退款启动消费者！");
        Properties properties = rocketMqConfig.getProperties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, TopicAndTag.GID_YYKSNORMAL_CANCEL_ORDER);
        // 集群消费模式
        properties.setProperty(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_CNBOOKING_ORDER_CANCEL, orderRefundListener);
        consumer.start();
        log.info("启动订单退款消费者成功！");
        return consumer;
    }

    /**
     * 会员卡支付成功监听
     *
     * @return
     */
    @Bean(name = "miniCardPaySuccessConsumer")
    public Consumer getMiniCardPaySuccessConsumer(MiniCardPaySuccessListener miniCardPaySuccessListener) {
        log.info("初始化会员卡支付成功启动消费者！");
        Properties properties = rocketMqConfig.getProperties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, TopicAndTag.GID_YYKSNORMAL_CARD_PAY_SUCCESS);
        // 集群消费模式
        properties.setProperty(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_CARD_PAY_SUCCESS, miniCardPaySuccessListener);
        consumer.start();
        log.info("启动会员卡支付成功消费者成功！");
        return consumer;
    }

//    /**
//     * 小黑鱼礼包取消监听
//     *
//     * @return
//     */
//    @Bean(name = "bigGiftCancleConsumer")
//    public Consumer getBigGiftCancleConsumer(BigGiftCancleListener bigGiftCancleListener) {
//        log.info("初始化小黑鱼礼包取消监听！");
//        Properties properties = rocketMqConfig.getProperties();
//        properties.setProperty(PropertyKeyConst.GROUP_ID, TopicAndTag.GID_YYKSDELAY_BIGGITORDERCHECK);
//        // 集群消费模式
//        properties.setProperty(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
//        Consumer consumer = ONSFactory.createConsumer(properties);
//        consumer.subscribe(TopicAndTag.TOPIC_YYKS_DELAY, TopicAndTag.TAG_BIGGIT_ORDER_CHECK, bigGiftCancleListener);
//        consumer.start();
//        log.info("启动小黑鱼礼包取消监听成功！");
//        return consumer;
//    }

}
